/*
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE
 * file distributed with this work for additional information regarding copyright ownership. The ASF licenses this file
 * to You under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the
 * License. You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
 * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
 * specific language governing permissions and limitations under the License.
 */
package com.swak.boot.config.dubbo.event;

import static java.util.concurrent.Executors.newSingleThreadExecutor;
import static org.springframework.util.ObjectUtils.containsElement;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.dubbo.common.lang.ShutdownHookCallbacks;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.event.ContextClosedEvent;
import org.springframework.context.event.SmartApplicationListener;
import org.springframework.util.ClassUtils;

import com.swak.reactivex.threads.SwakThreadFactory;

/**
 * Awaiting Non-Web Spring Boot {@link ApplicationListener}
 *
 * @since 2.7.0
 */
public class AwaitingNonWebApplicationListener implements SmartApplicationListener {

	private static final String[] WEB_APPLICATION_CONTEXT_CLASSES = new String[] {
			"com.swak.reactivex.context.ReactiveServerApplicationContext" };

	private static final Logger logger = LoggerFactory.getLogger(AwaitingNonWebApplicationListener.class);

	@SuppressWarnings("unchecked")
	private static final Class<? extends ApplicationEvent>[] SUPPORTED_APPLICATION_EVENTS = of(
			ApplicationReadyEvent.class, ContextClosedEvent.class);

	private static final AtomicBoolean awaited = new AtomicBoolean(false);
	private static final AtomicBoolean awaked = new AtomicBoolean(false);

	private static final Lock lock = new ReentrantLock();
	private static final Condition condition = lock.newCondition();

	private static final ExecutorService executorService = newSingleThreadExecutor(
			new SwakThreadFactory("Dubbo-Await.", false, new AtomicInteger(1)));

	@SuppressWarnings("unchecked")
	private static <T> T[] of(T... values) {
		return values;
	}

	@Override
	public boolean supportsEventType(Class<? extends ApplicationEvent> eventType) {
		return containsElement(SUPPORTED_APPLICATION_EVENTS, eventType);
	}

	@Override
	public boolean supportsSourceType(Class<?> sourceType) {
		return true;
	}

	@Override
	public void onApplicationEvent(ApplicationEvent event) {
		if (event instanceof ApplicationReadyEvent) {
			this.onApplicationReadyEvent((ApplicationReadyEvent) event);
		} else if (event instanceof ContextClosedEvent) {
			this.onApplicationClosedEvent(event);
		}
	}

	protected void onApplicationReadyEvent(ApplicationReadyEvent event) {

		final ConfigurableApplicationContext applicationContext = event.getApplicationContext();

		if (isWebApplication(applicationContext)) {
			return;
		}
		if (awaited.compareAndSet(false, true)) {
			await();
			releaseOnExit();
		}
	}

	protected void onApplicationClosedEvent(ApplicationEvent event) {
		this.release();
	}

	private void releaseOnExit() {
		ShutdownHookCallbacks.INSTANCE.addCallback(this::release);
	}

	private boolean isWebApplication(ApplicationContext applicationContext) {
		boolean webApplication = false;
		for (String contextClass : WEB_APPLICATION_CONTEXT_CLASSES) {
			if (isAssignable(contextClass, applicationContext.getClass(), applicationContext.getClassLoader())) {
				webApplication = true;
				break;
			}
		}
		return webApplication;
	}

	private static boolean isAssignable(String target, Class<?> type, ClassLoader classLoader) {
		try {
			return ClassUtils.resolveClassName(target, classLoader).isAssignableFrom(type);
		} catch (Throwable ex) {
			return false;
		}
	}

	protected void await() {

		// has been waited, return immediately
		if (awaked.get()) {
			return;
		}

		if (!executorService.isShutdown()) {
			executorService.execute(() -> executeMutually(() -> {
				while (!awaked.get()) {
					if (logger.isInfoEnabled()) {
						logger.info(" [Dubbo] Current Spring Boot Application is await...");
					}
					try {
						condition.await();
					} catch (InterruptedException e) {
						Thread.currentThread().interrupt();
					}
				}
			}));
		}
	}

	protected void release() {
		executeMutually(() -> {
			while (awaked.compareAndSet(false, true)) {
				if (logger.isInfoEnabled()) {
					logger.info(" [Dubbo] Current Spring Boot Application is about to shutdown...");
				}
				condition.signalAll();
				// @since 2.7.8 method shutdown() is combined into the method release()
				shutdown();
			}
		});
	}

	private void shutdown() {
		if (!executorService.isShutdown()) {
			// Shutdown executorService
			executorService.shutdown();
		}
	}

	private void executeMutually(Runnable runnable) {
		try {
			lock.lock();
			runnable.run();
		} finally {
			lock.unlock();
		}
	}

	@Override
	public int getOrder() {
		return LOWEST_PRECEDENCE;
	}
}
